package org.example.union;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/18 10:03
 */
public class FlinkConnect {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        environment.setParallelism(1);

        DataStreamSource<Tuple2<Integer,String>> s1 = environment
                .fromCollection(Arrays.asList(
                        Tuple2.of(1, "a1"),
                        Tuple2.of(1, "a2"),
                        Tuple2.of(2, "b"),
                        Tuple2.of(3, "c")
                ));

        DataStreamSource<Tuple3<Integer,String,Integer>> s2 = environment
                .fromCollection(Arrays.asList(
                        Tuple3.of(2, "bb", 1),
                        Tuple3.of(1, "aa1", 1),
                        Tuple3.of(1, "aa2", 2),
                        Tuple3.of(3, "cc", 1),
                        Tuple3.of(4, "cc", 1)
                        )
                );

        SingleOutputStreamOperator<String> process = s1.connect(s2)
                .keyBy(key -> key.f0,key -> key.f0)
                .process(new CoProcessFunction<Tuple2<Integer,String>, Tuple3<Integer,String,Integer>, String>() {
            @Override
            public void processElement1(Tuple2<Integer, String> value,
                                        CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx,
                                        Collector<String> out) throws Exception {
                out.collect(String.format("%s,--->%s,%s","processElement1",value.f0,value.f1));
            }

            @Override
            public void processElement2(Tuple3<Integer, String, Integer> value,
                                        CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx,
                                        Collector<String> out) throws Exception {
                out.collect(String.format("%s,--->%s,%s","processElement2",value.f0,value.f1));
            }
        });

        process.print();

        environment.execute();

    }

}
